package gateway

import (
	"context"
	"github.com/centrifugal/centrifuge"
	"github.com/liserc/opensocket-server/tool/apiresp"
	"github.com/liserc/opensocket-server/tool/jsonutil"
	"github.com/liserc/opensocket-server/tool/log"
	"time"
)

// RPCExtensionFunc ...
type RPCExtensionFunc func(c Client, e centrifuge.RPCEvent) (centrifuge.RPCReply, error)

// Handler ...
type Handler struct {
	node           *centrifuge.Node
	messageHandle  MessageHandle
	messageEncoder Encoder
	rpcExtension   map[string]RPCExtensionFunc
}

func NewHandler(node *centrifuge.Node) *Handler {
	return &Handler{
		node:           node,
		messageEncoder: NewJsonEncoder(),
		rpcExtension:   make(map[string]RPCExtensionFunc),
	}
}

func (h *Handler) SetMessageHandle(messageHandle MessageHandle) {
	h.messageHandle = messageHandle
}

func (h *Handler) SetRPCExtension(method string, fnc RPCExtensionFunc) {
	h.rpcExtension[method] = fnc
}

func (h *Handler) Setup() error {
	h.registerClientRPCFunc()

	var concurrency = 5
	var semaphore chan struct{}
	semaphore = make(chan struct{}, concurrency)

	h.node.OnConnecting(func(ctx context.Context, e centrifuge.ConnectEvent) (centrifuge.ConnectReply, error) {
		reply, err := h.OnClientConnecting(ctx, e)
		if err != nil {
			return centrifuge.ConnectReply{}, err
		}
		return reply, err
	})

	h.node.OnConnect(func(client *centrifuge.Client) {
		transport := client.Transport()
		log.CInfo(client.Context(), "user connected with protocol", "UserID", client.UserID(), "Name", transport.Name(), "Protocol", transport.Protocol())
		client.OnRPC(func(event centrifuge.RPCEvent, cb centrifuge.RPCCallback) {
			h.runConcurrentlyIfNeeded(client.Context(), concurrency, semaphore, func() {
				reply, err := h.OnRPC(client, event)
				cb(reply, err)
			})
		})
		client.OnMessage(func(event centrifuge.MessageEvent) {

		})
	})

	return nil
}

func (h *Handler) OnClientConnecting(ctx context.Context, evt centrifuge.ConnectEvent) (centrifuge.ConnectReply, error) {
	var req ConnectReq
	err := jsonutil.JsonUnmarshal(evt.Data, &req)
	if err != nil {
		log.ZError(ctx, "客户端连接参数错误", err)
		return centrifuge.ConnectReply{}, centrifuge.DisconnectBadRequest
	}
	log.CInfo(ctx, "客户端开始连接", "data", req)
	if len(req.Token) == 0 {
		log.ZError(ctx, "客户端连接令牌为空", nil)
		return centrifuge.ConnectReply{}, centrifuge.DisconnectInvalidToken
	}

	userID := req.Token
	credentials := &centrifuge.Credentials{
		UserID:   userID,
		ExpireAt: time.Now().Unix() + 3600, // Expire in 5 seconds.
	}

	return centrifuge.ConnectReply{
		ClientSideRefresh: true, // This is required to use client-side refresh.
		Credentials:       credentials,
	}, nil
}

func (h *Handler) OnRPC(c Client, e centrifuge.RPCEvent) (centrifuge.RPCReply, error) {
	if handler, ok := h.rpcExtension[e.Method]; ok {
		return handler(c, e)
	}
	return centrifuge.RPCReply{}, centrifuge.ErrorMethodNotFound
}

func (h *Handler) registerClientRPCFunc() {
	h.SetRPCExtension("GetSeq", func(c Client, event centrifuge.RPCEvent) (centrifuge.RPCReply, error) {
		return h.handleClientRPCFunc(c, event, h.messageHandle.GetSeq)
	})
	h.SetRPCExtension("SendMessage", func(c Client, event centrifuge.RPCEvent) (centrifuge.RPCReply, error) {
		return h.handleClientRPCFunc(c, event, h.messageHandle.SendMessage)
	})
}

func (h *Handler) handleClientRPCFunc(c Client, event centrifuge.RPCEvent, fnc MessageFunc) (centrifuge.RPCReply, error) {
	ctx := c.Context()
	message := event.Data
	var req RpcReq
	var str = string(message)
	err := h.messageEncoder.Decode(message, &req)
	if err != nil {
		log.ZError(ctx, "WS Client Decode Error", err)
		return centrifuge.RPCReply{}, err
	}
	log.ZDebug(ctx, "WS Client Request", "req", req, "str", str)

	result, err := fnc(ctx, &req)
	if err != nil {
		return centrifuge.RPCReply{}, err
	}

	errResp := apiresp.ParseError(err)
	resp := RpcResp{
		ReqIdentifier: req.ReqIdentifier,
		MsgIncr:       req.MsgIncr,
		OperationID:   req.OperationID,
		ErrCode:       errResp.ErrCode,
		ErrMsg:        errResp.ErrMsg,
		Data:          result,
	}
	log.ZDebug(ctx, "WS Client Response", "resp", resp)

	encodedBuf, err := h.messageEncoder.Encode(&resp)
	if err != nil {
		return centrifuge.RPCReply{}, err
	}
	return centrifuge.RPCReply{Data: encodedBuf}, nil
}

func (h *Handler) runConcurrentlyIfNeeded(ctx context.Context, concurrency int, semaphore chan struct{}, fn func()) {
	if concurrency > 1 {
		select {
		case <-ctx.Done():
			return
		case semaphore <- struct{}{}:
		}
		go func() {
			defer func() { <-semaphore }()
			fn()
		}()
	} else {
		fn()
	}
}
